package demo;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.halo.trade.data.sina.Level2Handler;
import com.boco.base.util.datafile.CDataFile;

/**
 * 默认的level2的处理器
 * 由它来将数据发送至文件或者数据库
 *     目前入库的格式为：
 *          股票代码，秒或毫秒，
 *          key,数据类型,ms,数据字符串
 *     入缓存的数据为：
 *          股票代码
 *          key,数据类型,ms,数据字符串
 *     
 * */
public class DefaultLevel2Handler implements Level2Handler{
	//数据队列
	protected ConcurrentLinkedQueue linkedQueue = new ConcurrentLinkedQueue<>();
	
	
	private CDataFile g_DataFile;
	
	public void start(String fileSavePath){
		g_DataFile = new CDataFile();
		g_DataFile.put_FileExtensionName("DAT");
		g_DataFile.put_FullFilePath(fileSavePath);
		g_DataFile.put_FileNamePrefix("STOCK");
		g_DataFile.put_CreateFileInterval(600);
		g_DataFile.put_NumberOfMaxRecord(1000000);
		
		new DoDealMsg().start();
	}
	
	@Override
	public void onLevel2(int fetch_type, Map<String, Object> datas)
	{
	     return;
	}

	@Override
	public void onMessage(String msg) {
		if(msg!=null){
			String[] msgList = msg.split("\n");
			for (String strMsg : msgList){
				this.linkedQueue.offer(strMsg);	
			}
		}
	}
	
	class DoDealMsg extends Thread{
		@Override
		public void run() {
			while(true){
				Object objMsg = linkedQueue.poll();
				//为空表示没有数据
				if (objMsg == null){
					try {
						sleep(100);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				else{
					doMsg((String)objMsg);
				}
			}
		}
	}

	public void doMsg(String msg) {
		try {
			if (msg == null) {
				return;
			}
			if (msg.equals("")) {
				return;
			}

			String[] msgKeyValue = msg.split("=");
			if (msgKeyValue.length < 2) {
				return;
			}
			// 第一位是股票代码，第二位是类型，如果没有，表示行情
			msgKeyValue[0] = msgKeyValue[0].replace("[pushmsg]", "");
			String[] stockkey = msgKeyValue[0].replace("2cn_", "").substring(2).split("_");
			// 获取时间戳
			// 2cn_sz002214=????,14:53:27
			// 2cn_sz002214_1=1137316|13:15:44.260|
			// 2cn_sz002214_0=1134205|13:14:59.310
			// 2cn_sh600406_orders=15:00:00.000,15:00:00.000
			// _i的数据没有。
			String[] tmpMsg = msgKeyValue[1].split("[,|]");
            if(tmpMsg.length<=1){
            	return;
            }
			String timeKey = tmpMsg[1];
			// 数据齐全了，可以写文件或和处理redis。

			String[] resultMsg = new String[4];
			resultMsg[0] = stockkey[0];
			if (stockkey.length == 2) {
				resultMsg[1] = stockkey[1];
			} else {
				resultMsg[1] = "hq";
				try {
					if(tmpMsg.length>=26){
						float sell = Float.parseFloat(tmpMsg[26]);
						float buy = Float.parseFloat(tmpMsg[46]);
						String sprateRate = String.format("%.4f", (buy - sell) / sell);
						msgKeyValue[1] += ("," + sprateRate);
					}
					else{
						return;
					}
				} catch (Exception ex) {
					ex.printStackTrace();
				}
			}
			resultMsg[2] = timeKey;
			resultMsg[3] = msgKeyValue[1];
			// 对数据做一下计算，放到redis里
			g_DataFile.writeString(
					String.format("%s,%s,%s,%s\n", resultMsg[0], resultMsg[1], resultMsg[2], resultMsg[3]));
			// System.out.println(String.format("%s,%s,%s,%s", resultMsg[0],
			// resultMsg[1], resultMsg[2], resultMsg[3]));
		} catch (Exception ex) {
			ex.printStackTrace();
			System.out.println(msg);
		}
	};
	
	public static void main(String[] args) throws InterruptedException {
		DefaultLevel2Handler handler = new DefaultLevel2Handler();
		handler.start( "/Users/toriycn/workspace/studio/codetest/test/");
		FileReader reader;
		try {
			reader = new FileReader("/Users/toriycn/workspace/studio/javadefault/oschina/sinav2/002214.txt");
			BufferedReader br = new BufferedReader(reader);
	        String str = null;
	        while((str = br.readLine()) != null) {
	    		handler.doMsg(str);
	        }
	        br.close();
	        reader.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
        Thread.sleep(1000*1000);
	}
}
